package org.example.kafkastudy.config;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.RoundRobinPartitioner;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.*;

import java.util.*;
import java.util.stream.Collectors;

@Configuration
public class KafkaConfig {

    @Autowired
    KafkaProperties kafkaProperties;

    @Bean
    public ProducerFactory<String, String> producerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getBootstrapServers()); // 可以直接使用 application.yaml 的配置
        configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); // 如果在 application.yaml 中配置了序列化方式，也可以通过 kafkaProperties 获取
        configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
//        configProps.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, RoundRobinPartitioner.class.getName()); // 配置分区策略

//        List<String> interceptors = new ArrayList<>();
//        interceptors.add(ImportantInterceptor.class.getName());
//        interceptors.add(SendTimeInterceptor.class.getName());
//        String interceptorsStr = interceptors.stream().collect(Collectors.joining(","));
//        configProps.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, interceptorsStr); // 配置拦截器
        return new DefaultKafkaProducerFactory<>(configProps);
    }

    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() { // 自己注册的 kafkaTemplate 会替换掉 Spring 容器中自动注入的 kafkaTemplate
        return new KafkaTemplate<>(producerFactory());
    }

    @Bean
    public ConsumerFactory<String, String> consumeFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getBootstrapServers()); // 可以直接使用 application.yaml 的配置
        configProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); // 如果在 application.yaml 中配置了序列化方式，也可以通过 kafkaProperties 获取
        configProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        return new DefaultKafkaConsumerFactory<>(configProps);
    }

    @Bean
    public KafkaListenerContainerFactory kafkaListenerContainerFactory(ConsumerFactory<String, String> consumeFactory, KafkaTemplate<String, String> kafkaTemplate) { // 参数注入
        ConcurrentKafkaListenerContainerFactory listenerContainerFactory = new ConcurrentKafkaListenerContainerFactory();
        listenerContainerFactory.setConsumerFactory(consumeFactory);
        listenerContainerFactory.setReplyTemplate(kafkaTemplate);
        return listenerContainerFactory;
    }

    @Bean
    public ConsumerFactory<String, String> orderConsumeFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getBootstrapServers()); // 可以直接使用 application.yaml 的配置
        configProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); // 如果在 application.yaml 中配置了序列化方式，也可以通过 kafkaProperties 获取
        configProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        configProps.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, OrderConsumeInterceptor.class.getName()); // 配置拦截器
        return new DefaultKafkaConsumerFactory<>(configProps);
    }

    @Bean
    public KafkaListenerContainerFactory orderKafkaListenerContainerFactory(ConsumerFactory<String, String> orderConsumeFactory) {
        ConcurrentKafkaListenerContainerFactory listenerContainerFactory = new ConcurrentKafkaListenerContainerFactory();
        listenerContainerFactory.setConsumerFactory(orderConsumeFactory);
        return listenerContainerFactory;
    }

    @Bean
    public ConsumerFactory<String, String> payConsumeFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getBootstrapServers()); // 可以直接使用 application.yaml 的配置
        configProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); // 如果在 application.yaml 中配置了序列化方式，也可以通过 kafkaProperties 获取
        configProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        configProps.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, PayConsumeInterceptor.class.getName()); // 配置拦截器
        return new DefaultKafkaConsumerFactory<>(configProps);
    }

    @Bean
    public KafkaListenerContainerFactory payKafkaListenerContainerFactory(ConsumerFactory<String, String> payConsumeFactory) {
        ConcurrentKafkaListenerContainerFactory listenerContainerFactory = new ConcurrentKafkaListenerContainerFactory();
        listenerContainerFactory.setConsumerFactory(payConsumeFactory);
        return listenerContainerFactory;
    }
}

